package io.kiki.sba.registry.server.metadata.bootstrap;

import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import com.google.common.base.Predicate;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.elector.Leader;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.metrics.ReporterUtils;
import io.kiki.sba.registry.metrics.TaskMetrics;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.server.metadata.bootstrap.config.MultiClusterMetaServerConfig;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.remoting.meta.LocalCentralControlExchanger;
import io.kiki.sba.registry.server.metadata.remoting.meta.MetaServerRenewService;
import io.kiki.sba.registry.server.shared.client.manager.ClientManagerService;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.store.api.elector.LeaderElector;
import io.kiki.sba.registry.store.api.meta.RecoverConfigRepository;
import org.apache.commons.lang.StringUtils;
import org.glassfish.jersey.server.ResourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

import javax.annotation.Resource;
import javax.ws.rs.Path;
import javax.ws.rs.ext.Provider;
import java.lang.annotation.Annotation;
import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


public class MetaServerBootstrap {

    private static final Logger logger = LoggerFactory.getLogger(MetaServerBootstrap.class);
    private final AtomicBoolean rpcServerForSessionStarted = new AtomicBoolean(false);
    private final AtomicBoolean rpcServerForDataStarted = new AtomicBoolean(false);
    private final AtomicBoolean rpcServerForMetaStarted = new AtomicBoolean(false);
    private final AtomicBoolean httpServerStarted = new AtomicBoolean(false);
    private final AtomicBoolean rpcServerForRemoteMetaStarted = new AtomicBoolean(false);
    private final AtomicBoolean schedulerStart = new AtomicBoolean(false);
    private final Retryer<Boolean> retryer = RetryerBuilder.<Boolean>newBuilder().retryIfException().retryIfResult(new Predicate<Boolean>() {
        @Override
        public boolean apply(Boolean input) {
            return !input;
        }
    }).withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(10)).build();
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private MultiClusterMetaServerConfig multiClusterMetaServerConfig;
    @Autowired
    private DataExchanger boltDataExchanger;
    @Autowired
    private DataExchanger jerseyDataExchanger;
    @Autowired
    private ExecutorManager executorManager;
    @Resource(name = "sessionServerHandlers")
    private Collection<AbstractServerHandler> sessionServerHandlers;
    @Resource(name = "dataServerHandlers")
    private Collection<AbstractServerHandler> dataServerHandlers;
    @Resource(name = "metaServerHandlers")
    private Collection<AbstractServerHandler> metaServerHandlers;
    @Resource(name = "remoteMetaServerHandlers")
    private Collection<AbstractServerHandler> remoteMetaServerHandlers;
    @Autowired
    private ResourceConfig jerseyResourceConfig;
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private LeaderElector leaderElector;
    @Autowired
    private MetaServerRenewService metaServerRenewService;
    @Autowired
    private LocalCentralControlExchanger localMetaExchanger;
    @Resource
    private ClientManagerService clientManagerService;
    @Autowired
    private RecoverConfigRepository recoverConfigRepository;
    private Server sessionServer;
    private Server dataServer;
    private Server metaServer;
    private Server httpServer;
    private Server remoteMetaServer;

    /**
     * Do initialized.
     */
    public void start() {
        try {
            logger.info("release properties: {}", ServerEnv.getReleaseProps());
            logger.info("the configuration items are as follows: " + serverConfig.toString());
            ReporterUtils.enablePrometheusDefaultExports();

            recoverConfigRepository.waitSynced();
            clientManagerService.waitSynced();
            openSessionRegisterServer();

            openDataRegisterServer();

            openMetaRegisterServer();

            openHttpServer();

            openRemoteMetaServer();

            // meta start loop to elector leader
            startElectorLoop();

            retryer.call(() -> {
                io.kiki.sba.registry.store.api.elector.Leader leader = leaderElector.getLeaderInfo();
                logger.info("[MetaBootstrap] retry elector meta leader: {}, epoch:{}", leader.getLeader(), leader.getEpoch());
                return !StringUtils.isEmpty(leader.getLeader());
            });

            startScheduler();

            // start renew node
            renewNode();
            retryer.call(() -> {
                Leader leader = localMetaExchanger.getLeader(serverConfig.getLocalDataCenter());
                logger.info("[MetaBootstrap] retry connect to meta leader: {}, client:{}", leader.getLeader(), localMetaExchanger.getClient());
                return StringUtils.isNotEmpty(leader.getLeader()) && localMetaExchanger.getClient() != null;
            });

            TaskMetrics.getInstance().registerBolt();
            io.kiki.sba.registry.store.api.elector.Leader leader = leaderElector.getLeaderInfo();

            logger.info("[MetaBootstrap] leader info: {}, [{}]", leader.getLeader(), leader.getEpoch());
            Runtime.getRuntime().addShutdownHook(new Thread(this::doStop));
        } catch (Throwable e) {
            logger.error("Bootstrap Meta Server got error!", e);
            throw new RuntimeException("Bootstrap Meta Server got error!", e);
        }
    }

    private void startElectorLoop() {
        leaderElector.change2Follow();
    }

    private void startScheduler() {

        try {
            if (schedulerStart.compareAndSet(false, true)) {
                executorManager.startScheduler();
                logger.info("Meta Scheduler started!");
            }
        } catch (Exception e) {
            schedulerStart.set(false);
            logger.error("Meta Scheduler start error!", e);
            throw new RuntimeException("Meta Scheduler start error!", e);
        }
    }

    public void destroy() {
        doStop();
    }

    private void doStop() {
        try {
            logger.info("{} Shutting down Meta Server..", new Date().toString());

            executorManager.stopScheduler();
            stopServer();

        } catch (Throwable e) {
            logger.error("Shutting down Meta Server error!", e);
        }
        logger.info("{} Meta server is now shutdown...", new Date().toString());
    }

    private void renewNode() {
        metaServerRenewService.startRenewer(serverConfig.getSchedulerHeartbeatIntervalSecs() * 1000);
    }

    private void openSessionRegisterServer() {
        try {
            if (rpcServerForSessionStarted.compareAndSet(false, true)) {
                sessionServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getSessionServerPort()), sessionServerHandlers.toArray(new ChannelHandler[sessionServerHandlers.size()]));

                logger.info("Open session node register server port {} success!", serverConfig.getSessionServerPort());
            }
        } catch (Exception e) {
            rpcServerForSessionStarted.set(false);
            logger.error("Open session node register server port {} error!", serverConfig.getSessionServerPort(), e);
            throw new RuntimeException("Open session node register server error!", e);
        }
    }

    private void openDataRegisterServer() {
        try {
            if (rpcServerForDataStarted.compareAndSet(false, true)) {
                dataServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getDataServerPort()), dataServerHandlers.toArray(new ChannelHandler[dataServerHandlers.size()]));

                logger.info("Open data node register server port {} success!", serverConfig.getDataServerPort());
            }
        } catch (Exception e) {
            rpcServerForDataStarted.set(false);
            logger.error("Open data node register server port {} error!", serverConfig.getDataServerPort(), e);
            throw new RuntimeException("Open data node register server error!", e);
        }
    }

    private void openMetaRegisterServer() {
        try {
            if (rpcServerForMetaStarted.compareAndSet(false, true)) {
                metaServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getMetaServerPort()), metaServerHandlers.toArray(new ChannelHandler[metaServerHandlers.size()]));

                logger.info("Open meta server port {} success!", serverConfig.getMetaServerPort());
            }
        } catch (Exception e) {
            rpcServerForMetaStarted.set(false);
            logger.error("Open meta server port {} error!", serverConfig.getMetaServerPort(), e);
            throw new RuntimeException("Open meta server error!", e);
        }
    }

    private void openHttpServer() {
        try {
            if (httpServerStarted.compareAndSet(false, true)) {
                bindResourceConfig();
                httpServer = jerseyDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getHttpServerPort()), new ResourceConfig[]{jerseyResourceConfig});
                logger.info("Open http server port {} success!", serverConfig.getHttpServerPort());
            }
        } catch (Exception e) {
            httpServerStarted.set(false);
            logger.error("Open http server port {} error!", serverConfig.getHttpServerPort(), e);
            throw new RuntimeException("Open http server error!", e);
        }
    }

    private void openRemoteMetaServer() {
        try {
            if (rpcServerForRemoteMetaStarted.compareAndSet(false, true)) {
                remoteMetaServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), multiClusterMetaServerConfig.getRemoteMetaServerPort()), remoteMetaServerHandlers.toArray(new ChannelHandler[remoteMetaServerHandlers.size()]));

                logger.info("Open remote meta server port {} success!", multiClusterMetaServerConfig.getRemoteMetaServerPort());
            }
        } catch (Exception e) {
            rpcServerForRemoteMetaStarted.set(false);
            logger.error("Open remote meta server port {} error!", multiClusterMetaServerConfig.getRemoteMetaServerPort(), e);
            throw new RuntimeException("Open remote meta server error!", e);
        }
    }

    private void bindResourceConfig() {
        registerInstances(Path.class);
        registerInstances(Provider.class);
    }

    private void registerInstances(Class<? extends Annotation> annotationType) {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(annotationType);
        if (beans != null && beans.size() > 0) {
            beans.forEach((beanName, bean) -> jerseyResourceConfig.registerInstances(bean));
        }
    }

    private void stopServer() {
        if (sessionServer != null && sessionServer.isOpen()) {
            sessionServer.close();
        }
        if (dataServer != null && dataServer.isOpen()) {
            dataServer.close();
        }
        if (metaServer != null && metaServer.isOpen()) {
            metaServer.close();
        }
        if (httpServer != null && httpServer.isOpen()) {
            httpServer.close();
        }

        if (remoteMetaServer != null && remoteMetaServer.isOpen()) {
            remoteMetaServer.isOpen();
        }
    }

    public boolean isRpcServerForSessionStarted() {
        return rpcServerForSessionStarted.get();
    }


    public boolean isRpcServerForDataStarted() {
        return rpcServerForDataStarted.get();
    }


    public boolean isRpcServerForMetaStarted() {
        return rpcServerForMetaStarted.get();
    }


    public boolean isHttpServerStarted() {
        return httpServerStarted.get();
    }

    public boolean isRpcServerForRemoteMetaStarted() {
        return rpcServerForRemoteMetaStarted.get();
    }
}
